package cn.itsnkkka.consumer;

import cn.itsnkkka.domain.dto.ARabbitMQObjectDTO;
import cn.itsnkkka.domain.dto.RabbitMQObjectDTO;
import cn.itsnkkka.domain.enums.RabbitMQMessageTypeEnum;
import cn.itsnkkka.utils.KryoUtils;
import cn.itsnkkka.utils.MQUtils;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author xuezipeng
 * @date 2023-04-23 12:52
 */
@Component
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class AConsumer implements IConsumer{

    private final ThreadPoolExecutor executorService;

    public AConsumer() {
        executorService = new ThreadPoolExecutor(
                THREAD_COUNT,
                THREAD_COUNT,
                0, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                r -> new Thread(r, "Executor-" + r.hashCode())
        );
    }

    private static final String QUEUE_NAME = "a_queue";

    @RabbitListener(queues = {QUEUE_NAME})
    @Override
    public void process(byte[] body, Channel channel) {
        try {
            channel.basicQos(QOS);
        } catch (IOException e) {
            log.error("程序异常：{}", e.getMessage());
            e.printStackTrace();
        }
        this.spin();
        executorService.execute(() -> {
            try{
                ARabbitMQObjectDTO rabbitMQObjectDTO = KryoUtils.readFromByteArray(body, ARabbitMQObjectDTO.class);
                // TODO 可以处理逻辑了；
                log.info("处理数据: {}", rabbitMQObjectDTO);
            }catch (Exception e){
                log.error("RabbitMQ Spring整体遇到问题: {}", e.getMessage());
                e.printStackTrace();
            }
        });

    }

    private void spin() {
        while (executorService.getActiveCount() == THREAD_COUNT) {
            log.warn("执行器线程active thread已满，开始自旋。线程池信息: {}", executorService);
            try {
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {
                log.error("执行器自旋睡眠过程出错!", e);
            }
        }
    }

}
